package cn.jly.flink.sideoutput;

import cn.jly.flink.entity.MetricEvent;
import cn.jly.flink.sideoutput.impl.MyRedisCommandContainerImpl;
import com.alibaba.fastjson.JSON;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;
import redis.clients.jedis.HostAndPort;

import java.util.HashSet;
import java.util.Map;

/**
 * @PackageName cn.jly.flink.sideoutput
 * @ClassName FlinkSideOutputDemo
 * @Description 侧输出流测试
 * @Author 姬岚洋
 * @Date 2021/1/15 下午 5:30
 */
public class FlinkSideOutputDemo {
    private static final String MACHINE_TAG_NAME = "machine";
    private static final String DOCKER_TAG_NAME = "docker";
    private static final String APPLICATION_TAG_NAME = "application";
    private static final String MIDDLEWARE_TAG_NAME = "middleware";

    public static void main(String[] args) throws Exception {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        /*
        要使用 Side Output 的话，你首先需要做的是定义一个 OutputTag 来标识 Side Output，
        代表这个 Tag 是要收集哪种类型的数据，如果是要收集多种不一样类型的数据，那么你就需要定义多种 OutputTag
         */
        final OutputTag<MetricEvent> machineTag = new OutputTag<MetricEvent>(MACHINE_TAG_NAME) {
        };
        final OutputTag<MetricEvent> dockerTag = new OutputTag<MetricEvent>(DOCKER_TAG_NAME) {
        };
        final OutputTag<MetricEvent> applicationTag = new OutputTag<MetricEvent>(APPLICATION_TAG_NAME) {
        };
        final OutputTag<MetricEvent> middlewareTag = new OutputTag<MetricEvent>(MIDDLEWARE_TAG_NAME) {
        };

        // 从rbb中读，并输出到不同的侧输出流
        final SingleOutputStreamOperator<MetricEvent> processStream = env.addSource(new RichSourceFunction<MetricEvent>() {
            private MyRedisCommandContainer redisCommandContainer = null;

            @Override
            public void open(Configuration parameters) throws Exception {
                super.open(parameters);
                final HashSet<HostAndPort> set = new HashSet<>(6);
                String ip = "192.168.172.101";
                set.add(new HostAndPort(ip, 7001));
                set.add(new HostAndPort(ip, 7002));
                set.add(new HostAndPort(ip, 7003));
                set.add(new HostAndPort(ip, 7004));
                set.add(new HostAndPort(ip, 7005));
                set.add(new HostAndPort(ip, 7006));
                redisCommandContainer = new MyRedisCommandContainerImpl(set);
            }

            @Override
            public void run(SourceContext<MetricEvent> ctx) throws Exception {
                String hashKey = "MetricEvent";
                final Map<String, String> map = redisCommandContainer.hgetAll(hashKey);
                for (Map.Entry<String, String> entry : map.entrySet()) {
                    if (StringUtils.isNotEmpty(entry.getValue())) {
                        ctx.collect(JSON.parseObject(entry.getValue(), MetricEvent.class));
                    }
                }
            }

            @Override
            public void cancel() {
                if (redisCommandContainer != null) {
                    redisCommandContainer.close();
                }
            }
        })
                .process(new ProcessFunction<MetricEvent, MetricEvent>() {
                    @Override
                    public void processElement(MetricEvent metricEvent, Context ctx, Collector<MetricEvent> collector) throws Exception {
                        if (StringUtils.isNotEmpty(metricEvent.getEventTag())) {
                            switch (metricEvent.getEventTag()) {
                                case MACHINE_TAG_NAME:
                                    ctx.output(machineTag, metricEvent);
                                    break;
                                case APPLICATION_TAG_NAME:
                                    ctx.output(applicationTag, metricEvent);
                                    break;
                                case DOCKER_TAG_NAME:
                                    ctx.output(dockerTag, metricEvent);
                                    break;
                                case MIDDLEWARE_TAG_NAME:
                                    ctx.output(middlewareTag, metricEvent);
                                    break;
                                default:
                                    collector.collect(metricEvent);
                                    break;
                            }
                        }
                    }
                });

        // 打印正常的输出流
        processStream.print("其他事件");

        // 获取侧输出流
        processStream.getSideOutput(machineTag).print("机器相关事件");
        processStream.getSideOutput(applicationTag).print("应用相关事件");
        processStream.getSideOutput(dockerTag).print("docker相关事件");
        processStream.getSideOutput(middlewareTag).print("中间件相关事件");

        env.execute("FlinkSideOutputDemo");
    }
}
